﻿using Confluent.Kafka;
using Microsoft.Extensions.Options;
using System;
using System.Collections.Generic;
using System.Text;

namespace EventBus.Kafka
{
    public class kafkaConnection : IkafkaConnection, IDisposable
    {
        private readonly KafkaOptions _options;
        private IProducer<string, byte[]> _producer;
        private IConsumer<string, byte[]> _consumer;
        public kafkaConnection(IOptions<KafkaOptions> options)
        {
            _options = options.Value;
        }

        /// <summary>
        /// 获取product实例
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <returns></returns>
        public IProducer<string, byte[]> GetProducer()
        {
            try
            {
                if (_producer != null)
                {
                    return _producer;
                }
                var conf = new ProducerConfig
                {
                    BootstrapServers = _options.BootstrapServers,
                    Acks = _options.Acks,
                    Partitioner = _options.Partitioner,
                };

                //获取product实例
                _producer = new ProducerBuilder<string, byte[]>(conf).Build();
                return _producer;
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
                throw;
            }
        }

        public IConsumer<string, byte[]> GetConsumer(string groupId) 
        {
            try
            {
                //if (_consumer != null)
                //{
                //    return _consumer;
                //}
                var conf = new ConsumerConfig
                {
                    GroupId = groupId, //消费者组Id，相同的Id，在同一个消费者组中
                    BootstrapServers = _options.BootstrapServers,
                    // Note: The AutoOffsetReset property determines the start offset in the event
                    // there are not yet any committed offsets for the consumer group for the
                    // topic/partitions of interest. By default, offsets are committed
                    // automatically, so in this example, consumption will only start from the
                    // earliest message in the topic 'my-topic' the first time you run the program.
                    EnableAutoCommit = _options.EnableAutoCommit, //禁止自动确认消息               
                    AutoOffsetReset = _options.AutoOffsetReset
                };
                _consumer = new ConsumerBuilder<string, byte[]>(conf).Build();
                return _consumer;
            }
            catch (Exception)
            {

                throw;
            }
        }

        /// <summary>
        /// 释放product实例
        /// </summary>
        public void Dispose()
        {
            if (_producer != null) 
            {
                _producer.Dispose();
            }

            if (_consumer != null)
            {
                _consumer.Dispose();
            }
        }
    }
}
